package com.raylu.day04;

import com.raylu.utils.ClickEvent;
import com.raylu.utils.ClickSource;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.username)
                .process(new MyProcess())
                .print();

        env.execute();
    }

    public static class MyProcess extends KeyedProcessFunction<String, ClickEvent, String> {
        // key: url
        // value: count
        private MapState<String, Integer> urlCount;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            urlCount = getRuntimeContext().getMapState(
                    new MapStateDescriptor<String, Integer>(
                            "url-count",
                            Types.STRING, // key的泛型
                            Types.INT // value的泛型
                    )
            );
        }

        @Override
        public void processElement(ClickEvent value, Context ctx, Collector<String> out) throws Exception {
            if (!urlCount.contains(value.url)) {
                urlCount.put(value.url, 1);
            } else {
                urlCount.put(value.url, urlCount.get(value.url) + 1);
            }

            StringBuilder result = new StringBuilder();
            result.append(value.username + " => {\n");
            for (String key : urlCount.keys()) {
                result.append("  " + key + " : " + urlCount.get(key) + ",\n");
            }
            result.append("}\n");
            out.collect(result.toString());
        }
    }
}
